package cn._51doit.live.udf;

import cn._51doit.live.pojo.DataBean;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class IsNewUserFunctionV3 extends KeyedProcessFunction<String, DataBean, DataBean> {

    private transient ValueState<Integer> uidState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("uid-state", Integer.class);
        uidState = getRuntimeContext().getState(stateDescriptor);
    }

    @Override
    public void processElement(DataBean bean, Context ctx, Collector<DataBean> out) throws Exception {

        Integer value = uidState.value();
        if (value == null) {
            value = 1;
            bean.setIsN(1); //设置为新用户
        }
        uidState.update(value);

        out.collect(bean);
    }
}
